package com.qdlc.p2p.common.core.queue.queue;

import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.concurrent.Future;

import org.apache.commons.beanutils.BeanUtils;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.context.ApplicationContext;

import com.lmax.disruptor.BatchEventProcessor;
import com.lmax.disruptor.BlockingWaitStrategy;
import com.lmax.disruptor.EventFactory;
import com.lmax.disruptor.EventHandler;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.SequenceBarrier;
import com.lmax.disruptor.SingleThreadedClaimStrategy;
import com.lmax.disruptor.WorkHandler;
import com.qdlc.p2p.common.core.queue.AbstractQueueHandler;
import com.qdlc.p2p.common.core.queue.QueueHandlers;
import com.qdlc.p2p.common.core.queue.ValueEvent;
import com.qdlc.p2p.common.core.queue.event.AbstractValueEvent;

/**
 * 使用Disruptor框架实现
 * @author linya 2015-12-31
 */
public class DisruptorQueueHandler extends AbstractQueueHandler implements InitializingBean{

	//RingBuffer容量
	private static final int BUFFER_SIZE = 1024;
	
	//事件工厂，生成空的EmptyVauleEvent实例
	public final static EventFactory<ValueEvent> EVENT_FACTORY = new EventFactory<ValueEvent>() {
		public ValueEvent newInstance() {
			return new EmptyValueEvent();
		}
	};
	
	//创建环形缓存器
	private final RingBuffer<ValueEvent> ringBuffer = new RingBuffer<ValueEvent>(EVENT_FACTORY,
			new SingleThreadedClaimStrategy(BUFFER_SIZE), new BlockingWaitStrategy());

	//序列屏障
	private final SequenceBarrier sequenceBarrier = ringBuffer.newBarrier();

	//事件批量处理器
	private final BatchEventProcessor<ValueEvent> batchEventProcessor = new BatchEventProcessor<ValueEvent>(ringBuffer,
			sequenceBarrier, new TaskEventHandler());
	
	
	@Override
	public void afterPropertiesSet() throws Exception {
		ringBuffer.setGatingSequences(batchEventProcessor.getSequence());
	}
	
	@Override
	public Future<ValueEvent> submit(ValueEvent event) {
		put(event);
		Future<ValueEvent> future = scheduledTask(new BatchEventCallable(batchEventProcessor, event));
		//事件批量回调器
		return future;
	}

	@Override
	public List<Future<ValueEvent>> submit(Collection<? extends ValueEvent> events) {
		List<Future<ValueEvent>> futureList = new ArrayList<Future<ValueEvent>>();
		for(ValueEvent event: events){
			futureList.add(submit(event));
		}
		return futureList;
	}

	@SuppressWarnings("unused")
	@Override
	public void put0(ValueEvent event) {
		long sequence = ringBuffer.next();
		try {
			if (event != null) {
				boolean capacity = ringBuffer.hasAvailableCapacity(BUFFER_SIZE/10);
				if (capacity) {
					// TODO: RingBuffer快满时，需要处理！
				} else {
					//将事件对象进行更换（偷天换日），因为ringBuffer中存放的是EmptyVauleEvent实例，
					//需要更换为入队的ValueEvent（只是更改了引用）
					ValueEvent e = ringBuffer.get(sequence);
					e = (ValueEvent) BeanUtils.cloneBean(event);
				}
			}
		} catch (Exception e) {
			e.printStackTrace();
		}finally{
			ringBuffer.publish(sequence);
		}
	}

	
	@Override
	protected void postApplicationContext(ApplicationContext context) {
		QueueHandlers.initPostAppliactionContext(context);
	}

	@Override
	protected boolean isEmpty() {
		return ringBuffer.hasAvailableCapacity(ringBuffer.getBufferSize());
	}

	@Override
	protected void doRun() {
		taskExecutor.submit(batchEventProcessor);
	}

	
	/**
	 * 空的ValueEvent事件，由于RingBuffer中预先指定事件对象
	 * @author linya
	 */
	@SuppressWarnings("serial")
	public static class EmptyValueEvent extends AbstractValueEvent{
		
	}
	
	//保存线程所执行的ValueEvent事件
	ThreadLocal<ValueEvent> threadLocal = new ThreadLocal<ValueEvent>();
	
	public class TaskEventHandler implements EventHandler<ValueEvent>, WorkHandler<ValueEvent>{
		@Override
		public void onEvent(ValueEvent event, long sequence, boolean endOfBatch)
				throws Exception {
			this.onEvent(event);
		}

		@Override
		public void onEvent(ValueEvent event) throws Exception {
			process(event);
			threadLocal.set(event);	//保存ValueEvent
		}
	}

	public class BatchEventCallable implements EventTasker{
		
		BatchEventProcessor<ValueEvent> processor;
		
		ValueEvent valueEvent;
		
		BatchEventCallable(BatchEventProcessor<ValueEvent> processor, ValueEvent valueEvent){
			this.processor = processor;
			this.valueEvent = valueEvent;
		}
		
		@Override
		public ValueEvent call() throws Exception {
			processor.run();
			return threadLocal.get();
		}

		@Override
		public ValueEvent getValueEvent() {
			return valueEvent;
		}
	}

}
